/*
* Copyright (C) 2010-2012 Magnus Skjegstad
* 
* This program is free software; you can redistribute it and/or
* modify it under the terms of the GNU General Public License
* as published by the Free Software Foundation; either version 2
* of the License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program; if not, write to the Free Software
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
* 
* The latest version of this library can be downloaded from http://mist-pubsub.googlecode.com
* 
*/
package com.skjegstad.mist2.demos;


import com.skjegstad.mist2.MistCoreManager;
import com.skjegstad.mist2.datatypes.DataElement;
import com.skjegstad.mist2.exceptions.MistException;
import java.io.BufferedInputStream;
import java.io.IOException;
import java.net.InetAddress;
import java.net.URL;
import java.net.UnknownHostException;
import java.util.Arrays;
import java.util.Random;
import java.util.concurrent.*;

/**
 *
 * @author Magnus Skjegstad
 */
public class TestDataProducer {
public static void main(String[] argv) throws UnknownHostException, IOException, MistException, InterruptedException {
        final Random r = new Random();
        MistCoreManager mist = new MistCoreManager();
        //mist.enableBroadcast(InetAddress.getByName("255.255.255.255"), 5579);
        mist.enableMulticast(InetAddress.getByName("224.0.0.1"), 5579);

        mist.setDebug(true);
        mist.start();

        final int totalBytesPerSecond = 55000;
        final int packetsPerSecond = 1;
        final int packetSize = totalBytesPerSecond / packetsPerSecond;
        

        final BlockingQueue<byte[]> dataQueue = new ArrayBlockingQueue<byte[]>(100);


        long start = System.currentTimeMillis();

        // stream from file
        //final FileInputStream inStream = new FileInputStream(new File("Nordkalotten365.E05.1024x576.H264.NRK.mp4"));

        // stream from std in
        //final BufferedInputStream inStream = new BufferedInputStream(System.in);

        // stream from url
        URL url = new URL("http://lyd.nrk.no/nrk_radio_p3_mp3_h");
        final BufferedInputStream inStream = new BufferedInputStream(url.openStream());


        ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);
        executor.scheduleAtFixedRate(new Runnable(){
            public void run() {
                byte[] data = new byte[packetSize];
                try {
                    //r.nextBytes(data); int read = data.length; // fill array with random data
                    int read = inStream.read(data); // remove this line and replace with random to generate random traffic

                    if (read != packetSize) {
                        data = Arrays.copyOf(data, read);
                    }
                } catch (Exception ex) {
                    ex.printStackTrace(System.err);
                    return;
                }
                dataQueue.add(data);
            }
        }, 100, 1000 / packetsPerSecond, TimeUnit.MILLISECONDS);

        long sentSoFar = 0;
        
        while (true && !Thread.interrupted()) {
            byte[] input = dataQueue.take();

            sentSoFar = sentSoFar + input.length;
                        
            DataElement e = mist.publish("no.ffi.stream.test", input, 30000);
            
            double bw = sentSoFar / ((System.currentTimeMillis() - start) / 1000.0);
            
            System.err.println("Produced message #" + e.getSeqNr() + ", " + input.length + " bytes (bw=" + bw + " bytes per second)");
        }
        
        mist.stop();
        
    }
}
